package com.foreveross.fetl.common.hbase.client;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;

import com.foreveross.fetl.common.vo.AgentInfo;
import com.foreveross.fetl.common.vo.PlaneInfo;

public class HBasePlaneinfoHelper {

	private static Logger log = Logger.getLogger(HBasePlaneinfoHelper.class);
	
	public static List<PlaneInfo> transformHBaseResultToPlaneinfo(
			ResultScanner scan) {
		List<PlaneInfo> result = new ArrayList<PlaneInfo>();
		for (Result r : scan) {
			PlaneInfo planeInfo = transformHBaseResultToPlaneinfo(r);
			result.add(planeInfo);
		}
		return result;
	}
	
	/**
	 * 将HBase中的一行记录转换为Planeinfo对象
	 * @param r
	 * @return
	 */
	public static PlaneInfo transformHBaseResultToPlaneinfo(Result r) {
	
		PlaneInfo planeInfo = new PlaneInfo();
		List<AgentInfo> agents = new ArrayList<AgentInfo>();
		Map<String, AgentInfo> agentMap = new HashMap<String, AgentInfo>();
		try {
			for (KeyValue kv : r.raw()) {
				if ("planeinfo".equals(new String(kv.getFamily(), "utf-8"))) {
					// 创建planeInfo对象
					String property = new String(kv.getQualifier(), "utf-8");
					try {
						Method method = planeInfo.getClass().getMethod(
								"set"
										+ property.substring(0, 1)
												.toUpperCase()
										+ property.substring(1),
								String.class);
						method.invoke(planeInfo, new String(kv.getValue(),
								"utf-8"));
					} catch (Exception ex) {
						//log.error(ex.getMessage());
					}
				}

				// 组建agentInfo对象
				else if ("agentInfos".equals(new String(kv.getFamily(),
						"utf-8"))) {
					String str = new String(kv.getQualifier(), "utf-8")
							.split("_")[1];
					if (!agentMap.containsKey(str)) {
						agentMap.put(str, new AgentInfo());
					}
					if (agentMap.keySet().contains(
							new String(kv.getQualifier(), "utf-8")
									.split("_")[1])) {
						String property = new String(kv.getQualifier(),
								"utf-8").split("_")[0];
						try{
							Method method = agentMap
								.get(new String(kv.getQualifier(), "utf-8").split("_")[1]).getClass()
								.getMethod("set"+ property.substring(0, 1).toUpperCase()+ property.substring(1),String.class);
							method.invoke(agentMap.get(new String(kv
									.getQualifier(), "utf-8").split("_")[1]),
									new String(kv.getValue(), "utf-8"));
						} catch (Exception ex) {
							//log.error(ex.getMessage());
						}
					}
				}
			}
			agents.addAll(agentMap.values());
			planeInfo.setAgentInfos(agents);
		} catch (Exception e) {
			//log.error(e.getMessage());
		}
		return planeInfo;	
	}
}
